package org.restaurantexample.ws;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.cassandra.thrift.IndexOperator;

import org.sjsusearch.www.restaurantservice.Restaurant;
import org.sjsusearch.www.restaurantservice.RestaurantList;

import me.prettyprint.cassandra.model.BasicColumnDefinition;
import me.prettyprint.cassandra.model.BasicColumnFamilyDefinition;
import me.prettyprint.cassandra.serializers.IntegerSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.ThriftCfDef;
import me.prettyprint.cassandra.service.template.ColumnFamilyResult;
import me.prettyprint.cassandra.service.template.ColumnFamilyTemplate;
import me.prettyprint.cassandra.service.template.ColumnFamilyUpdater;
import me.prettyprint.cassandra.service.template.IndexedSlicesPredicate;
import me.prettyprint.cassandra.service.template.ThriftColumnFamilyTemplate;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ColumnIndexType;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
import me.prettyprint.hector.api.exceptions.HectorException;
import me.prettyprint.hector.api.factory.HFactory;

/*
 * Restaurant DAO class which creates and stores the Data into the database
 */
public class RestaurantDAO {
	 
	 //Database Name and Table Name
	 public static String keyspaceName = "Lab2_275", columnFamilyName = "Restaurant";
	 
	 public static Cluster cluster;
	 
	 //Column Names
	 public static String column1 = "Restaurant_Id", column2 = "Restaurant_Name", column3 ="Cuisine",
	 			column4 ="Restaurant_Rating",column5="Latitude",column6="Longitude", column7="Restaurant_Address",
	 			column8 = "Restaurant_All";
	 
	 /*
	  * Function to create the database
	  */
	private static void createSchema()
   {
		try
		{
			BasicColumnFamilyDefinition columnFamilyDefinition = new BasicColumnFamilyDefinition();
	        columnFamilyDefinition.setKeyspaceName(keyspaceName);
	        columnFamilyDefinition.setName(columnFamilyName);    
	        columnFamilyDefinition.setDefaultValidationClass("UTF8Type");
	        columnFamilyDefinition.setKeyValidationClass(ComparatorType.UTF8TYPE.getClassName());
	        columnFamilyDefinition.setComparatorType(ComparatorType.UTF8TYPE);
	
	        BasicColumnDefinition columnDefinition = new BasicColumnDefinition();
	        
	        //Id Index
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column1));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.INTEGERTYPE.getClassName());
	        columnDefinition.setIndexName(column1+"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	
	        //Name Index
	        columnDefinition = new BasicColumnDefinition();
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column2));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.UTF8TYPE.getClassName());
	        columnDefinition.setIndexName(column2+"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	  
	        //Cuisine Index
			columnDefinition = new BasicColumnDefinition();
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column3));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.UTF8TYPE.getClassName());
	        columnDefinition.setIndexName(column3 +"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	        
	        //All Index
			columnDefinition = new BasicColumnDefinition();
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column8));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.INTEGERTYPE.getClassName());
	        columnDefinition.setIndexName(column8 +"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	       
	        //Rating Index
	        columnDefinition = new BasicColumnDefinition();
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column4));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.UTF8TYPE.getClassName());
	        columnDefinition.setIndexName(column4+"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	        
	        ColumnFamilyDefinition cfDef = new ThriftCfDef(columnFamilyDefinition);
	        
	        KeyspaceDefinition keyspaceDefinition = HFactory.createKeyspaceDefinition(keyspaceName, 
	            		"SimpleStrategy", 1, Arrays.asList(cfDef));
	        
	        cluster.addKeyspace(keyspaceDefinition);    	
		}
		catch(Exception e)
		{
			System.out.println("Exception occured: " + e.getMessage());
		}
   }
	
	 /*
	  * Function to create the database
	  */
	private static void updateSchema()
	{
		try
		{
			BasicColumnFamilyDefinition columnFamilyDefinition = new BasicColumnFamilyDefinition();
	        columnFamilyDefinition.setKeyspaceName(keyspaceName);
	        columnFamilyDefinition.setName(columnFamilyName);    
	        columnFamilyDefinition.setDefaultValidationClass("UTF8Type");
	        columnFamilyDefinition.setKeyValidationClass(ComparatorType.UTF8TYPE.getClassName());
	        columnFamilyDefinition.setComparatorType(ComparatorType.UTF8TYPE);
	
	        BasicColumnDefinition columnDefinition = new BasicColumnDefinition();
	        
	        //Id Index
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column1));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.INTEGERTYPE.getClassName());
	        columnDefinition.setIndexName(column1+"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	
	        //Name Index
	        columnDefinition = new BasicColumnDefinition();
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column2));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.UTF8TYPE.getClassName());
	        columnDefinition.setIndexName(column2+"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	  
	        //Cuisine Index
			columnDefinition = new BasicColumnDefinition();
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column3));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.UTF8TYPE.getClassName());
	        columnDefinition.setIndexName(column3 +"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	       
	        //All Index
			columnDefinition = new BasicColumnDefinition();
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column8));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.INTEGERTYPE.getClassName());
	        columnDefinition.setIndexName(column8 +"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	        
	        //Rating Index
	        columnDefinition = new BasicColumnDefinition();
	        columnDefinition.setName(StringSerializer.get().toByteBuffer(column4));
	        columnDefinition.setIndexType(ColumnIndexType.KEYS);
	        columnDefinition.setValidationClass(ComparatorType.UTF8TYPE.getClassName());
	        columnDefinition.setIndexName(column4+"_index");
	        columnFamilyDefinition.addColumnDefinition(columnDefinition);
	        
	        ColumnFamilyDefinition cfDef = new ThriftCfDef(columnFamilyDefinition);
	        
	        cluster.addColumnFamily(cfDef);    	
		}
		catch(Exception e)
		{
			System.out.println("Exception occured: " + e.getMessage());
		}
  }
	
	/*
	 * Function to create the database
	 */
	private static void createDatabase()
	{
		try
		{
			cluster = HFactory.getOrCreateCluster("Test Cluster","localhost:9160");
			KeyspaceDefinition keyspaceDef = cluster.describeKeyspace(keyspaceName);
	
			if(keyspaceDef == null)
			{
				  	createSchema();
			}
			else
			{
				updateSchema();
			}
							
			System.out.println("Resturant Database created successfully!!!!");
		}
		catch(Exception e)
		{
			System.out.println("Exception occured: " + e.getMessage());
		}
	}
	
	/*
	 * Function to insert values into the database
	 */
	private static void insertData()
	{
		KeyspaceDefinition keyspaceDef = cluster.describeKeyspace(keyspaceName);

		if(keyspaceDef == null)
		{
			createDatabase();
		}
		
		 Keyspace keyspace = HFactory.createKeyspace(keyspaceName, cluster);
		 
		//Insert Values into the table
		try 
		{
			for(int i = 1; i<51; i++)
			{
				ColumnFamilyTemplate<String, String> template = 
	                new ThriftColumnFamilyTemplate<String, String>(keyspace,columnFamilyName, 
	                                                     StringSerializer.get(), StringSerializer.get());
				
				ColumnFamilyUpdater<String, String> updater = template.createUpdater("row" + i);
				updater.setInteger(column1, i);
			
				if(i == 1)
				{
					updater.setString(column2, "Acapulco");
					updater.setString(column3, "Mexican");
					updater.setString(column4, "4.5");
					updater.setString(column5, "37.20.21.N");
					updater.setString(column6, "121.53.38.W");
					updater.setString(column7, "1299,Lawerence Expressway, San Jose, CA ");
				}				
				if(i == 2)
				{
					updater.setString(column2, "Amarin Thai");
					updater.setString(column3, "Thai");
					updater.setString(column4, "5.0");
					updater.setString(column5, "36.20.21.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "174-176, Castro Street, San Jose,CA");
				}
				if(i == 3)
				{
					updater.setString(column2, "Amber");
					updater.setString(column3, "Indian");
					updater.setString(column4, "5.0");
					updater.setString(column5, "37.20.21.N");
					updater.setString(column6, "121.53.38.W");
					updater.setString(column7, "2290, West ElCamino, San Jose, CA");
					
				}
				if(i == 4)
				{
					updater.setString(column2, "Arka");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.0");
					updater.setString(column5, "36.19.20.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "725 S, Fair Oaks Ave, San Jose, CA");
				}
				if(i == 5)
				{
					updater.setString(column2, "Banana Leaf");
					updater.setString(column3, "Singaporean");
					updater.setString(column4, "5.0");
					updater.setString(column5, "36.20.21.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "182, Ranch Drive, San Jose, CA");
				}
				if(i == 6)
				{
					updater.setString(column2, "Bangalore Cafe");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "36.19.20.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "3077, El Camino Real, San Jose, CA");
				}
				if(i == 7)
				{
					updater.setString(column2, "Bare Foot");
					updater.setString(column3, "Coffee Shop");
					updater.setString(column4, "4.5");
					updater.setString(column5, "36.19.20.N");
					updater.setString(column6, "121.52.53.W");
					updater.setString(column7, "5237, Stevens Creek Blvd, SanJose, CA");
				}
				if(i == 8)
				{
					updater.setString(column2, "Barracuda");
					updater.setString(column3, "Japanese");
					updater.setString(column4, "4.5");
					updater.setString(column5, "37.20.21.N");
					updater.setString(column6, "121.53.38.W");
					updater.setString(column7, "124, Castro Street, San Jose, CA");
				}
				if(i == 9)
				{
					updater.setString(column2, "Bombay Garden");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "35.19.18.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "3701, El Camino Real, San Jose, CA");
				}
				if(i == 10)
				{
					updater.setString(column2, "Cantankerous Fish");
					updater.setString(column3, "Sea Food");
					updater.setString(column4, "4.0");
					updater.setString(column5, "33.21.22.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "420, Castro Street, San Jose, CA");
				}
				if(i == 11)
				{
					updater.setString(column2, "Chef Lee");
					updater.setString(column3, "Chinese");
					updater.setString(column4, "4.0");
					updater.setString(column5, "33.21.22.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "1115, Tasman Drive, San Jose, CA");
				}
				if(i == 12)
				{
					updater.setString(column2, "CheloKababi");
					updater.setString(column3, "Persian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "37.20.21.N");
					updater.setString(column6, "121.53.38.W");
					updater.setString(column7, "1236, South Wolfe Road, San Jose, CA");
				}
				if(i == 13)
				{
					updater.setString(column2, "Chipotle");
					updater.setString(column3, "Mexican");
					updater.setString(column4, "5.0");
					updater.setString(column5, "36.22.25.N");
					updater.setString(column6, "120.52.53.W");
					updater.setString(column7, "2400, Charelston Road, San Jose, CA");
				}
				if(i == 14)
				{
					updater.setString(column2, "Dasaprakash");
					updater.setString(column3, "Indian");
					updater.setString(column4, "3.5");
					updater.setString(column5, "36.20.21.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "2636, HomeStead Road, San Jose, CA");
				}
				if(i == 15)
				{
					updater.setString(column2, "DD's");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "33.21.22.N");
					updater.setString(column6, "120.52.53.W");
					updater.setString(column7, "341, Lafayette Street, San Jose, CA");
				}
				if(i == 16)
				{
					updater.setString(column2, "Evvia");
					updater.setString(column3, "Greek");
					updater.setString(column4, "5.0");
					updater.setString(column5, "36.19.20.N");
					updater.setString(column6, "120.52.53.W");
					updater.setString(column7, "420, Emerson Street, San Jose, CA");
				}
				if(i == 17)
				{
					updater.setString(column2, "Hangen");
					updater.setString(column3, "Chinese");
					updater.setString(column4, "5.0");
					updater.setString(column5, "35.19.18.N");
					updater.setString(column6, "122.52.53.W");
					updater.setString(column7, "134 Castro Street, San Jose, CA");
				}
				if(i == 18)
				{
					updater.setString(column2, "Junoon");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "36.19.21.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "150, University Avenue, San Jose, CA");
				}
				if(i == 19)
				{
					updater.setString(column2, "King of Krung Siam");
					updater.setString(column3, "Siamese");
					updater.setString(column4, "4.5");
					updater.setString(column5, "36.22.25.N");
					updater.setString(column6, "122.52.53.W");
					updater.setString(column7, "194, Castro street, San Jose, CA");
				}
				if(i == 20)
				{
					updater.setString(column2, "Kirin");
					updater.setString(column3, "Japanese");
					updater.setString(column4, "5.0");
					updater.setString(column5, "34.33.35.N");
					updater.setString(column6, "120.52.53.W");
					updater.setString(column7, "2700, Yulupa Avenue, San Jose, CA");
				}
				if(i == 21)
				{
					updater.setString(column2, "La Paloma");
					updater.setString(column3, "Mexican");
					updater.setString(column4, "4.5");
					updater.setString(column5, "34.33.35.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "2280, El Camino Real, San Jose, CA");
				}
				if(i == 22)
				{
					updater.setString(column2, "Mediterranean Grill");
					updater.setString(column3, "Mediterranean");
					updater.setString(column4, "4.5");
					updater.setString(column5, "36.22.25.N");
					updater.setString(column6, "122.52.53.W");
					updater.setString(column7, "650, Castro Street, San Jose, CA");
				}
				if(i == 23)
				{
					updater.setString(column2, "Momoya Sushi");
					updater.setString(column3, "Japanese");
					updater.setString(column4, "5.0");
					updater.setString(column5, "33.21.22.N");
					updater.setString(column6, "121.50.54.W");
					updater.setString(column7, "570, N Shoreline Blvd, San Jose, CA");
				}
				if(i == 24)
				{
					updater.setString(column2, "Okayama");
					updater.setString(column3, "Japanese");
					updater.setString(column4, "5.0");
					updater.setString(column5, "36.20.21.N");
					updater.setString(column6, "121.50.54.W");
					updater.setString(column7, "565, N 6th Street, San Jose, CA");
				}
				if(i == 25)
				{
					updater.setString(column2, "Olive Garden");
					updater.setString(column3, "Italian");
					updater.setString(column4, "5.0");
					updater.setString(column5, "35.24.27.N");
					updater.setString(column6, "120.52.53.W");
					updater.setString(column7, "2515, El Camino Real, San Jose, CA");
				}
				if(i == 26)
				{
					updater.setString(column2, "P F Chang's");
					updater.setString(column3, "Chinese");
					updater.setString(column4, "4.0");
					updater.setString(column5, "34.33.35.N");
					updater.setString(column6, "121.50.54.W");
					updater.setString(column7, "900, Stanford Shopping Center, San Jose, CA");
				}
				if(i == 27)
				{
					updater.setString(column2, "Panera Bread");
					updater.setString(column3, "Italian");
					updater.setString(column4, "4.0");
					updater.setString(column5, "37.20.21.N");
					updater.setString(column6, "121.53.38.W");
					updater.setString(column7, "20807, Stevens Creek Blvd, San Jose, CA");
				}
				if(i == 28)
				{
					updater.setString(column2, "Pasta Pomodoro");
					updater.setString(column3, "Italian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "35.24.27.N");
					updater.setString(column6, "122.52.53.W");
					updater.setString(column7, "378, Santana Row, San Jose, CA");
				}
				if(i == 29)
				{
					updater.setString(column2, "Pasta?");
					updater.setString(column3, "Italian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "35.19.18.N");
					updater.setString(column6, "121.54.55.W");
					updater.setString(column7, "160, Castro Street, San Jose, CA");
				}
				if(i == 30)
				{
					updater.setString(column2, "Sakoon");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "36.22.25.N");
					updater.setString(column6, "122.53.54.W");
					updater.setString(column7, "357, Castro Street, San Jose, CA");
				}
				if(i == 31)
				{
					updater.setString(column2, "Salang Pass");
					updater.setString(column3, "Italian");
					updater.setString(column4, "5.0");
					updater.setString(column5, "35.19.18.N");
					updater.setString(column6, "121.54.55.W");
					updater.setString(column7, "37462, Fremont Blvd, San Jose, CA");
				}
				if(i == 32)
				{
					updater.setString(column2, "Shana Thai");
					updater.setString(column3, "Thai");
					updater.setString(column4, "4.0");
					updater.setString(column5, "33.21.22.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "311, Moffet Blvd, San Jose, CA");
				}
				if(i == 33)
				{
					updater.setString(column2, "Shezan");
					updater.setString(column3, "Pakistani");
					updater.setString(column4, "4.0");
					updater.setString(column5, "35.19.18.N");
					updater.setString(column6, "122.52.53.W");
					updater.setString(column7, "216, Castro Street, San Jose, CA");
				}
				if(i == 34)
				{
					updater.setString(column2, "Shiva's");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.0");
					updater.setString(column5, "36.20.21.N");
					updater.setString(column6, "121.53.54.W");
					updater.setString(column7, "800, California Street, San Jose, CA");
				}
				if(i == 35)
				{
					updater.setString(column2, "Sweet Tomatoes");
					updater.setString(column3, "Salad & Bakery");
					updater.setString(column4, "4.5");
					updater.setString(column5, "34.33.35.N");
					updater.setString(column6, "121.52.37.W");
					updater.setString(column7, "1040, Grant Road, San Jose, CA");
				}
				if(i == 36)
				{
					updater.setString(column2, "Taste Buds");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "37.20.21.N");
					updater.setString(column6, "121.53.38.W");
					updater.setString(column7, "673, Grape Avenue, San Jose, CA");
				}
				if(i == 37)
				{
					updater.setString(column2, "Temptations");
					updater.setString(column3, "Indo-Chinese");
					updater.setString(column4, "4.0");
					updater.setString(column5, "35.24.27.N");
					updater.setString(column6, "121.56.57.W");
					updater.setString(column7, "288, Castro Street, San Jose, CA");
				}
				if(i == 38)
				{
					updater.setString(column2, "Thai Pepper");
					updater.setString(column3, "Thai");
					updater.setString(column4, "5.0");
					updater.setString(column5, "36.22.25.N");
					updater.setString(column6, "122.53.54.W");
					updater.setString(column7, "103, East Fremont Avenue,San Jose, CA");
				}
				if(i == 39)
				{
					updater.setString(column2, "Tide");
					updater.setString(column3, "Italian");
					updater.setString(column4, "5.0");
					updater.setString(column5, "33.21.22.N");
					updater.setString(column6, "122.52.53.W");
					updater.setString(column7, "954, Villa Street, San Jose, CA");
				}
				if(i == 40)
				{
					updater.setString(column2, "Tumeric");
					updater.setString(column3, "Indian");
					updater.setString(column4, "4.0");
					updater.setString(column5, "35.24.27.N");
					updater.setString(column6, "122.53.54.W");
					updater.setString(column7, "141, South Murphy Avenue, San Jose, CA");
				}
				if(i == 41)
				{
					updater.setString(column2, "Willow Tree");
					updater.setString(column3, "Chinese");
					updater.setString(column4, "4.0");
					updater.setString(column5, "35.19.18.N");
					updater.setString(column6, "121.50.54.W");
					updater.setString(column7, "633, Court Street, San Jose, CA");
				}
				if(i == 42)
				{
					updater.setString(column2, "Xanh");
					updater.setString(column3, "Vietnamese");
					updater.setString(column4, "5.0");
					updater.setString(column5, "35.24.27.N");
					updater.setString(column6, "121.56.57.W");
					updater.setString(column7, "110, Castro Street, San Jose, CA");
				}
				if(i == 43)
				{
					updater.setString(column2, "Zeni");
					updater.setString(column3, "Ethiopian");
					updater.setString(column4, "4.5");
					updater.setString(column5, "36.20.21.N");
					updater.setString(column6, "121.54.55.W");
					updater.setString(column7, "1320, Saratoga Avenue, San Jose, CA");
				}
				if(i == 44)
				{
					updater.setString(column2, "Zitune");
					updater.setString(column3, "Moroccan");
					updater.setString(column4, "4.0");
					updater.setString(column5, "34.33.35.N");
					updater.setString(column6, "121.56.57.W");
					updater.setString(column7, "325, Main Street, San Jose, CA");
				}
				if(i == 45)
				{
					updater.setString(column2, "Zucca Ristorante");
					updater.setString(column3, "Mediterranean");
					updater.setString(column4, "4.5");
					updater.setString(column5, "35.24.27.N");
					updater.setString(column6, "122.53.54.W");
					updater.setString(column7, "186, Castro Street, San Jose, CA");
				}
				if(i == 46)
				{
					updater.setString(column2, "Samovar");
					updater.setString(column3, "Russian");
					updater.setString(column4, "4.0");
					updater.setString(column5, "35.19.18.N");
					updater.setString(column6, "121.54.55.W");
					updater.setString(column7, "1077, Independence Avenue, San Jose, CA");
				}
				if(i == 47)
				{
					updater.setString(column2, "Sunny Bowl");
					updater.setString(column3, "Korean");
					updater.setString(column4, "4.0");
					updater.setString(column5, "34.33.35.N");
					updater.setString(column6, "121.50.54.W");
					updater.setString(column7, "1477, Plymouth Street, San Jose, CA");
				}
				if(i == 48)
				{
					updater.setString(column2, "Pastis");
					updater.setString(column3, "French");
					updater.setString(column4, "4.0");
					updater.setString(column5, "36.20.21.N");
					updater.setString(column6, "121.56.57.W");
					updater.setString(column7, "South California Avenue, San Jose, CA");
				}
				if(i == 49)
				{
					updater.setString(column2, "Pintxo");
					updater.setString(column3, "Spanish");
					updater.setString(column4, "4.5");
					updater.setString(column5, "36.22.25.N");
					updater.setString(column6, "121.54.55.W");
					updater.setString(column7, "Google Campus, 1600 AmphiTheatre,San Jose, CA ");
				}
				if(i == 50)
				{
					updater.setString(column2, "Bona");
					updater.setString(column3, "Polish");
					updater.setString(column4, "4.5");
					updater.setString(column5, "37.20.21.N");
					updater.setString(column6, "121.53.38.W");
					updater.setString(column7, "651, Oak Groove Avenue,San Jose, CA");
				}
				updater.setInteger(column8, 0);
				try 
				{
				    template.update(updater);
				} 
				catch (HectorException e) {
				    // do something ...
				}

			}
			System.out.println("Data added successfully to the database");			
		}
		catch(Exception e)
		{
			System.out.println("Exception occured: " + e.getMessage());
		}
				
	}
	
	
   public static void main(String[] args) throws Exception 
   { 
		//Create Schema if database does not exists
		createDatabase();
		
		//Insert Data to Database
		insertData();
				
		//Get All Data
		//List<Restaurant> allData = RestaurantDAO.getAllRestaurantData();
		
		//System.out.println("No Of Rows Retrived : " + allData.size());

		//Get Restaurant Data by Name 
		//Restaurant restaurantdataByName = getRestaurantByName("Amber");
		//System.out.println("No Of Rows Retrived : " + restaurantdataByName.getCuisine());
		
		//Get Restaurant Data by Cusine
		//List<Restaurant> restaurantDataByCusine =  getRestaurantByCusine("Indian");
		//System.out.println("No Of Rows Retrived : " + restaurantDataByCusine.size());
		
		//Get Restaurant Data by Rating
		//List<Restaurant> restaurantDataByRating =  getRestaurantByRating(5.0);
		//System.out.println("No Of Rows Retrived : " + restaurantDataByRating.size());
		
		if(cluster != null)
		{
			cluster.getConnectionManager().shutdown();
		}	
   } 
	
	/*
	 * Get all the restaurant data stored in the database
	 */
	public RestaurantList getAllRestaurantData() throws ServiceUnavailableException
	{
		try
		{
			RestaurantList result = new RestaurantList();
			cluster = HFactory.getOrCreateCluster("Test Cluster","localhost:9160");
			
			Keyspace keyspace = HFactory.createKeyspace(keyspaceName, cluster);
			ColumnFamilyTemplate<String, String> template = 
	            new ThriftColumnFamilyTemplate<String, String>(keyspace,
	                                                           columnFamilyName, 
	                                                           StringSerializer.get(),        
	                                                           StringSerializer.get());
		
			IndexedSlicesPredicate<String, String, Integer> predicate = new IndexedSlicesPredicate<String, String, Integer>
										(StringSerializer.get(), StringSerializer.get(), IntegerSerializer.get());
			
			predicate.startKey("");
			
			//Column 8 Dummy column added to get all the values
			predicate.addExpression(column8, IndexOperator.EQ, 0);
					
			predicate.addExpression(column1, IndexOperator.GT, 0);
			
			ColumnFamilyResult<String,String> res1 = template.queryColumns(predicate);
					
			//result = new ArrayList<Restaurant>();
			
			while(res1.hasResults())
			{
				Restaurant restaurantData = new Restaurant();
				restaurantData.setRestaurant_id(res1.getInteger(column1));
				restaurantData.setRestaurant_name(res1.getString(column2));
			    restaurantData.setCuisine(res1.getString(column3));
				restaurantData.setRestaurant_rating(Double.parseDouble(res1.getString(column4)));
			    restaurantData.setRestaurant_latitude(res1.getString(column5));
			    restaurantData.setRestaurant_longitude(res1.getString(column6));
			    restaurantData.setRestaurant_location(res1.getString(column7));
			    result.addRestaurantList(restaurantData);
			    
			    if(res1.hasNext())
			    {
			    	res1.next();
			    }
			    else
			    {
			    	break;
			    }
			}	    
	        return result;
		}
		catch(Exception e)
		{
			System.err.println("Exception occured in DAO layer: " + e.getMessage());
			throw new ServiceUnavailableException();
		}
	}
	
	/*
	 * Get all Details of a particular restaurant
	 */
	public Restaurant getRestaurantByName(String restaurantName) throws org.restaurantexample.ws.ServiceUnavailableException
	{
		try
		{
			Restaurant restaurantData = new Restaurant();
			cluster = HFactory.getOrCreateCluster("Test Cluster","localhost:9160");
			
			Keyspace keyspace = HFactory.createKeyspace(keyspaceName, cluster);
			ColumnFamilyTemplate<String, String> template = 
	            new ThriftColumnFamilyTemplate<String, String>(keyspace,
	                                                           columnFamilyName, 
	                                                           StringSerializer.get(),        
	                                                           StringSerializer.get());
		
			IndexedSlicesPredicate<String, String, String> predicate = new IndexedSlicesPredicate<String, String, String>
										(StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
			
			predicate.startKey("");
			predicate.addExpression(column2, IndexOperator.EQ, restaurantName);
			ColumnFamilyResult<String,String> res1 = template.queryColumns(predicate);
			
			restaurantData.setRestaurant_id(res1.getInteger(column1));
			restaurantData.setRestaurant_rating(Double.parseDouble(res1.getString(column4)));		
			restaurantData.setRestaurant_name(res1.getString(column2));
		    restaurantData.setCuisine(res1.getString(column3));
		    restaurantData.setRestaurant_latitude(res1.getString(column5));
		    restaurantData.setRestaurant_longitude(res1.getString(column6));
		    restaurantData.setRestaurant_location(res1.getString(column7));
		    
	        return restaurantData;
		}
		catch(Exception e)
		{
			System.err.println("Exception occured in DAO layer: " + e.getMessage());
			throw new ServiceUnavailableException();
		}
	}
	
	/*
	 * Get all restaurants which match the Cusine
	 */
	public RestaurantList getRestaurantByCusine(String cusineName) throws ServiceUnavailableException
	{
		try
		{
			RestaurantList result = new RestaurantList();
			cluster = HFactory.getOrCreateCluster("Test Cluster","localhost:9160");
			
			Keyspace keyspace = HFactory.createKeyspace(keyspaceName, cluster);
			ColumnFamilyTemplate<String, String> template = 
	            new ThriftColumnFamilyTemplate<String, String>(keyspace,
	                                                           columnFamilyName, 
	                                                           StringSerializer.get(),        
	                                                           StringSerializer.get());
		
			IndexedSlicesPredicate<String, String, String> predicate = new IndexedSlicesPredicate<String, String, String>
										(StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
			
			predicate.startKey("");
			predicate.addExpression(column3, IndexOperator.EQ, cusineName);
			ColumnFamilyResult<String,String> res1 = template.queryColumns(predicate);
					
			//result = new ArrayList<Restaurant>();
			
			while(res1.hasResults())
			{
				Restaurant restaurantData = new Restaurant();
				restaurantData.setRestaurant_id(res1.getInteger(column1));
				restaurantData.setRestaurant_name(res1.getString(column2));
			    restaurantData.setCuisine(res1.getString(column3));
				restaurantData.setRestaurant_rating(Double.parseDouble(res1.getString(column4)));
			    restaurantData.setRestaurant_latitude(res1.getString(column5));
			    restaurantData.setRestaurant_longitude(res1.getString(column6));
			    restaurantData.setRestaurant_location(res1.getString(column7));
			    result.addRestaurantList(restaurantData);
			    
			    if(res1.hasNext())
			    {
			    	res1.next();
			    }
			    else
			    {
			    	break;
			    }
			}	    
	        return result;
		}
		catch(Exception e)
		{
			System.err.println("Exception occured in DAO layer: " + e.getMessage());
			throw new ServiceUnavailableException();
		}
	}
	
	/*
	 * Get all restaurant detail which has specific rating
	 */
	public RestaurantList getRestaurantByRating(double value) throws ServiceUnavailableException
	{
		try
		{
			RestaurantList result = new RestaurantList();
			cluster = HFactory.getOrCreateCluster("Test Cluster","localhost:9160");
			
			Keyspace keyspace = HFactory.createKeyspace(keyspaceName, cluster);
			ColumnFamilyTemplate<String, String> template = 
	            new ThriftColumnFamilyTemplate<String, String>(keyspace,
	                                                           columnFamilyName, 
	                                                           StringSerializer.get(),        
	                                                           StringSerializer.get());
		
			IndexedSlicesPredicate<String, String, String> predicate = new IndexedSlicesPredicate<String, String, String>
										(StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
			
			predicate.startKey("");
			predicate.addExpression(column4, IndexOperator.EQ, String.valueOf(value));
			ColumnFamilyResult<String,String> res1 = template.queryColumns(predicate);
					
			//result = new ArrayList<Restaurant>();
			
			while(res1.hasResults())
			{
				Restaurant restaurantData = new Restaurant();
				restaurantData.setRestaurant_id(res1.getInteger(column1));
				restaurantData.setRestaurant_name(res1.getString(column2));
			    restaurantData.setCuisine(res1.getString(column3));
				restaurantData.setRestaurant_rating(Double.parseDouble(res1.getString(column4)));
			    restaurantData.setRestaurant_latitude(res1.getString(column5));
			    restaurantData.setRestaurant_longitude(res1.getString(column6));
			    restaurantData.setRestaurant_location(res1.getString(column7));
			    result.addRestaurantList(restaurantData);
			    
			    if(res1.hasNext())
			    {
			    	res1.next();
			    }
			    else
			    {
			    	break;
			    }
			}	    
	        return result;	
		}
		catch(Exception e)
		{
			System.err.println("Exception occured in DAO layer: " + e.getMessage());
			throw new ServiceUnavailableException();
		}
	}
}
